package mapreduce

import (
	"encoding/json"
	"io"
	"log"
	"os"
	"sort"
)

func doReduce(
	jobName string, // the name of the whole MapReduce job
	reduceTask int, // which reduce task this is
	outFile string, // write the output here
	nMap int, // the number of map tasks that were run ("M" in the paper)
	reduceF func(key string, values []string) string,
) {
	//
	// doReduce manages one reduce task: it should read the intermediate
	// files for the task, sort the intermediate key/value pairs by key,
	// call the user-defined reduce function (reduceF) for each key, and
	// write reduceF's output to disk.
	//
	// You'll need to read one intermediate file from each map task;
	// reduceName(jobName, m, reduceTask) yields the file
	// name from map task m.
	//
	// Your doMap() encoded the key/value pairs in the intermediate
	// files, so you will need to decode them. If you used JSON, you can
	// read and decode by creating a decoder and repeatedly calling
	// .Decode(&kv) on it until it returns an error.
	//
	// You may find the first example in the golang sort package
	// documentation useful.
	//
	// reduceF() is the application's reduce function. You should
	// call it once per distinct key, with a slice of all the values
	// for that key. reduceF() returns the reduced value for that key.
	//
	// You should write the reduce output as JSON encoded KeyValue
	// objects to the file named outFile. We require you to use JSON
	// because that is what the merger than combines the output
	// from all the reduce tasks expects. There is nothing special about
	// JSON -- it is just the marshalling format we chose to use. Your
	// output code will look something like this:
	//
	// enc := json.NewEncoder(file)
	// for key := ... {
	// 	enc.Encode(KeyValue{key, reduceF(...)})
	// }
	// file.Close()
	//
	// Your code here (Part I).
	//

	merge_bucket :=[]KeyValue{}
	//gather intermediate files,nMap satrt from zero
	for i:=0;i<nMap;i++{
		interm_filename := reduceName(jobName,i,reduceTask)
		fd,err:=os.Open(interm_filename)
		if err!=nil{
			log.Fatalf("Open Intermediate file error%s",interm_filename)
		}
		//read file content
		dec := json.NewDecoder(fd)
		for{
			//Single Line
			var kv KeyValue
			if err:=dec.Decode(&kv);err==io.EOF{
				break;
			}else if err!=nil{
				log.Fatal(err)
			}
			merge_bucket = append(merge_bucket,kv)
		}
	}
	//sort is needed
	sort.Sort(ByKey(merge_bucket))

	values := []string{}
	final_res := []KeyValue{}
	if len(merge_bucket)>0{
		values = append(values,merge_bucket[0].Value)
	}
	for startLoc:=1;startLoc<len(merge_bucket);startLoc++ {
		if merge_bucket[startLoc].Key!=merge_bucket[startLoc-1].Key{
			res:=reduceF(merge_bucket[startLoc-1].Key,values)
			final_res = append(final_res,KeyValue{merge_bucket[startLoc-1].Key,res})
			values = []string{}
			values = append(values,merge_bucket[startLoc].Value)
		}else{
			values = append(values,merge_bucket[startLoc].Value)
		}
	}
	//clear the values array
	if len(values)>0 {
		res:=reduceF(merge_bucket[len(merge_bucket)-1].Key, values)
		final_res = append(final_res,KeyValue{merge_bucket[len(merge_bucket)-1].Key,res})
	}
	//store value to disk
	fd,err:=os.Create(outFile+"test")
	if err!=nil{
		log.Fatalf("Failed to create outputFile")
	}
	enc:=json.NewEncoder(fd)
	for _, kv := range final_res {
		err := enc.Encode(&kv)
		if (err != nil) {
			log.Fatalf("JSON encode error%s", kv)
		}
	}
	err = os.Rename(outFile+"test", outFile)
	if err!=nil{
		log.Fatalf("OS rename Error!%s",outFile)
	}
	fd.Close()
}
